/**
 * 上海中赢金融信息服务有限公司
 * Copyright (c) 2017-2027 Chinazyjr,Inc.All Rights Reserved.
 */

package com.sys.midware.memcached;

import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import net.rubyeye.xmemcached.MemcachedClient;
import net.rubyeye.xmemcached.MemcachedClientBuilder;
import net.rubyeye.xmemcached.XMemcachedClientBuilder;
import net.rubyeye.xmemcached.command.BinaryCommandFactory;
import net.rubyeye.xmemcached.impl.KetamaMemcachedSessionLocator;
import net.rubyeye.xmemcached.utils.AddrUtil;

/**
 * <b>ClassName:</b> MemCachedCluster.java <br/>
 * <b>Description:</b> TODO <br/>
 * <b>Date:</b> 2015年11月18日 下午6:36:31<br/>
 * 
 * @author mobing
 * @version
 */
public class MemCachedCluster {
    private static class GetTask implements Callable {

        public MemCachedResult call() throws Exception {
            if (client == null)
                throw new IllegalArgumentException("memcached client is null");
            if (collection == null || collection.isEmpty())
                throw new IllegalArgumentException("key is null");
            long start = System.currentTimeMillis();
            Map results = new LinkedHashMap();
            List keys = new LinkedList(collection);
            int size = keys.size();
            for (int index = 0; index < size; index += batchSize) {
                Map map = client.get(keys.subList(index, Math.min(index + batchSize, size)));
                results.putAll(map);
            }

            long end = System.currentTimeMillis();
            return new MemCachedResult(0, end - start, results);
        }

//        public volatile Object call() throws Exception {
//            return call();
//        }

        private MemcachedClient client;

        private Collection collection;

        private int batchSize;

        public GetTask(MemcachedClient client, Collection collection) {
            batchSize = 100;
            this.client = client;
            this.collection = collection;
        }
    }

    private static class MemCachedResult {

        public int getCode() {
            return code;
        }

        public void setCode(int code) {
            this.code = code;
        }

        public long getUsedTime() {
            return usedTime;
        }

        public void setUsedTime(long usedTime) {
            this.usedTime = usedTime;
        }

        public Map getData() {
            return data;
        }

        public void setData(Map data) {
            this.data = data;
        }

        private int code;

        private long usedTime;

        private Map data;

        public MemCachedResult(int code, long usedTime) {
            this.code = code;
            this.usedTime = usedTime;
        }

        public MemCachedResult(int code, long usedTime, Map data) {
            this.code = code;
            this.usedTime = usedTime;
            this.data = data;
        }
    }

    private static class PutTask implements Callable {

        public MemCachedResult call() throws Exception {
            if (client == null)
                throw new IllegalArgumentException("memcached client is null");
            if (map == null || map.isEmpty())
                throw new IllegalArgumentException("key is null");
            long start = System.currentTimeMillis();
            for (Iterator iterator = map.entrySet().iterator(); iterator.hasNext();) {
                java.util.Map.Entry entry = (java.util.Map.Entry) iterator.next();
                String key = (String) entry.getKey();
                Object value = entry.getValue();
                if (key != null)
                    client.setWithNoReply(key, expiredTimeInSeconds, value);
            }

            long end = System.currentTimeMillis();
            return new MemCachedResult(0, end - start);
        }

//        public volatile Object call() throws Exception {
//            return call();
//        }

        private MemcachedClient client;

        private Map map;

        private int expiredTimeInSeconds;

        public PutTask(MemcachedClient client, Map map, int expiredTimeInSeconds) {
            this.client = client;
            this.map = map;
            this.expiredTimeInSeconds = expiredTimeInSeconds;
        }
    }

    private static class RemoveTask implements Callable {

        public MemCachedResult call() throws Exception {
            if (client == null)
                throw new IllegalArgumentException("memcached client is null");
            if (collection == null || collection.isEmpty())
                throw new IllegalArgumentException("key is null");
            long start = System.currentTimeMillis();
            for (Iterator iterator = collection.iterator(); iterator.hasNext();) {
                String key = (String) iterator.next();
                if (key != null)
                    client.deleteWithNoReply(key);
            }

            long end = System.currentTimeMillis();
            return new MemCachedResult(0, end - start);
        }

//        public volatile Object call() throws Exception {
//            return call();
//        }

        private MemcachedClient client;

        private Collection collection;

        private int batchSize;

        public RemoveTask(MemcachedClient client, Collection collection) {
            batchSize = 100;
            this.client = client;
            this.collection = collection;
        }
    }

    public MemCachedCluster() {
        es = Executors.newCachedThreadPool();
        clusters = new LinkedHashMap();
    }

    public static MemCachedCluster create() {
        return new MemCachedCluster();
    }

    public void addCluster(String memcachedServers) throws IOException {
        clusters.put(memcachedServers, createMemcachedClient(memcachedServers));
    }

    public Map get(Collection keys) throws InterruptedException, ExecutionException {
        CompletionService cs;
        if (keys == null || keys.isEmpty())
            return Collections.emptyMap();
        Collection clients = clusters.values();
        cs = new ExecutorCompletionService(es);
        GetTask task;
        for (Iterator iterator = clients.iterator(); iterator.hasNext(); cs.submit(task)) {
            MemcachedClient client = (MemcachedClient) iterator.next();
            task = new GetTask(client, keys);
        }

        MemCachedResult result;
        Future fs = cs.take();
        result = (MemCachedResult) fs.get();
        if (result == null)
            return Collections.emptyMap();
        try {
            return result.getData();
        } catch (Exception e) {
            logger.error((new StringBuilder("memcached cluster get error:")).append(e.getMessage()).toString(), e);
        }
        return Collections.emptyMap();
    }

    public Object get(String key) throws InterruptedException, ExecutionException {
        if (key == null)
            return null;
        Map map = get(((Collection) (Arrays.asList(new String[] { key }))));
        if (map == null || map.isEmpty())
            return null;
        else
            return map.get(key);
    }

    public void put(String key, Object value, int expiredTimeInSeconds) {
        if (key == null) {
            return;
        } else {
            Map map = new LinkedHashMap();
            map.put(key, value);
            put(map, expiredTimeInSeconds);
            return;
        }
    }

    public void put(Map map, int expiredTimeInSeconds) {
        if (map == null || map.isEmpty())
            return;
        Collection clients = clusters.values();
        CompletionService cs = new ExecutorCompletionService(es);
        PutTask task;
        for (Iterator iterator = clients.iterator(); iterator.hasNext(); cs.submit(task)) {
            MemcachedClient client = (MemcachedClient) iterator.next();
            task = new PutTask(client, map, expiredTimeInSeconds);
        }

        for (int i = 0; i < clients.size(); i++) {
            Future future;
            try {
                future = cs.take();
            } catch (Exception e) {
                logger.error((new StringBuilder("memcached cluster put error:")).append(e.getMessage()).toString(), e);
            }
        }

    }

    public void remove(Collection keys) {
        if (keys == null || keys.isEmpty())
            return;
        Collection clients = clusters.values();
        CompletionService cs = new ExecutorCompletionService(es);
        RemoveTask task;
        for (Iterator iterator = clients.iterator(); iterator.hasNext(); cs.submit(task)) {
            MemcachedClient client = (MemcachedClient) iterator.next();
            task = new RemoveTask(client, keys);
        }

        for (int i = 0; i < clients.size(); i++) {
            Future future;
            try {
                future = cs.take();
            } catch (Exception e) {
                logger.error((new StringBuilder("memcached cluster remove error:")).append(e.getMessage()).toString(),
                        e);
            }
        }

    }

    public void remove(String key) {
        if (key == null) {
            return;
        } else {
            remove(((Collection) (Arrays.asList(new String[] { key }))));
            return;
        }
    }

    public void shutDown() throws IOException {
        Collection clients = clusters.values();
        MemcachedClient client;
        for (Iterator iterator = clients.iterator(); iterator.hasNext(); client.shutdown())
            client = (MemcachedClient) iterator.next();

        if (es != null)
            es.shutdown();
        // throw exception;
        if (es != null)
            es.shutdown();
        return;
    }

    private MemcachedClient createMemcachedClient(String servers) throws IOException {
        MemcachedClientBuilder builder = new XMemcachedClientBuilder(AddrUtil.getAddresses(servers));
        builder.setSessionLocator(new KetamaMemcachedSessionLocator());
        builder.setCommandFactory(new BinaryCommandFactory());
        builder.setConnectTimeout(1000L);
        MemcachedClient memcachedClient = builder.build();
        memcachedClient.setTimeoutExceptionThreshold(2147483647);
        return memcachedClient;
    }

    private static final Logger logger = LoggerFactory
            .getLogger(MemCachedCluster.class);

    private ExecutorService es;

    private Map clusters;

    private static final long DEFAULT_CONNECTION_TIMEOUT = 1000L;
}
